package org.mx.dal.utils.impl;

import java.io.IOException;
import java.io.InputStream;
import java.lang.reflect.Field;
import java.nio.file.Files;
import java.nio.file.Paths;
import java.security.KeyManagementException;
import java.security.KeyStore;
import java.security.KeyStoreException;
import java.security.NoSuchAlgorithmException;
import java.security.cert.Certificate;
import java.security.cert.CertificateException;
import java.security.cert.CertificateFactory;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.function.BiConsumer;

import javax.net.ssl.SSLContext;

import org.apache.http.HttpHost;
import org.apache.http.auth.AuthScope;
import org.apache.http.auth.UsernamePasswordCredentials;
import org.apache.http.client.CredentialsProvider;
import org.apache.http.impl.client.BasicCredentialsProvider;
import org.apache.http.impl.nio.reactor.IOReactorConfig;
import org.apache.http.ssl.SSLContextBuilder;
import org.apache.http.ssl.SSLContexts;
import org.elasticsearch.client.RestClient;
import org.mx.ClassUtils;
import org.mx.DigestUtils;
import org.mx.StringUtils;
import org.mx.dal.Pagination;
import org.mx.dal.annotation.ElasticField;
import org.mx.dal.annotation.ElasticIndex;
import org.mx.dal.annotation.ElasticIndexAlias;
import org.mx.dal.entity.BaseDict;
import org.mx.dal.entity.ElasticBaseEntity;
import org.mx.dal.entity.ElasticGeoPointBaseEntity;
import org.mx.dal.entity.GeoPoint;
import org.mx.dal.entity.GeoShapePolygon;
import org.mx.dal.entity.PO;
import org.mx.dal.error.UserInterfaceDalErrorException;
import org.mx.dal.error.UserInterfaceDalErrorException.DalErrors;
import org.mx.dal.service.GeneralAccessor;
import org.mx.dal.service.GeneralAccessor.ConditionGroup;
import org.mx.dal.service.GeneralAccessor.ConditionTuple;
import org.mx.dal.service.GeneralAccessor.RecordOrder.OrderType;
import org.mx.dal.service.GeneralAccessor.RecordOrderGroup;
import org.mx.dal.utils.ElasticConfigBean;
import org.mx.dal.utils.ElasticLowLevelUtil;
import org.mx.dal.utils.ElasticUtil;
import org.mx.dal.utils.bean.IndicesInfoBean;
import org.mx.dal.utils.bean.NodeInfoBean;
import org.mx.error.UserInterfaceSystemErrorException;
import org.mx.error.UserInterfaceSystemErrorException.SystemErrors;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.alibaba.fastjson.JSONObject;

import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch._types.DistanceUnit;
import co.elastic.clients.elasticsearch._types.ElasticsearchException;
import co.elastic.clients.elasticsearch._types.FieldSort;
import co.elastic.clients.elasticsearch._types.FieldValue;
import co.elastic.clients.elasticsearch._types.GeoDistanceSort;
import co.elastic.clients.elasticsearch._types.GeoDistanceType;
import co.elastic.clients.elasticsearch._types.GeoLocation;
import co.elastic.clients.elasticsearch._types.GeoShapeRelation;
import co.elastic.clients.elasticsearch._types.LatLonGeoLocation;
import co.elastic.clients.elasticsearch._types.SortOptions;
import co.elastic.clients.elasticsearch._types.SortOrder;
import co.elastic.clients.elasticsearch._types.mapping.DynamicMapping;
import co.elastic.clients.elasticsearch._types.mapping.Property;
import co.elastic.clients.elasticsearch._types.query_dsl.BoolQuery;
import co.elastic.clients.elasticsearch._types.query_dsl.GeoShapeFieldQuery;
import co.elastic.clients.elasticsearch._types.query_dsl.Query;
import co.elastic.clients.elasticsearch._types.query_dsl.QueryBuilders;
import co.elastic.clients.elasticsearch.core.CountRequest;
import co.elastic.clients.elasticsearch.core.DeleteRequest;
import co.elastic.clients.elasticsearch.core.GetRequest;
import co.elastic.clients.elasticsearch.core.GetResponse;
import co.elastic.clients.elasticsearch.core.IndexRequest;
import co.elastic.clients.elasticsearch.core.IndexResponse;
import co.elastic.clients.elasticsearch.core.SearchRequest;
import co.elastic.clients.elasticsearch.core.SearchResponse;
import co.elastic.clients.elasticsearch.core.UpdateRequest;
import co.elastic.clients.elasticsearch.core.UpdateResponse;
import co.elastic.clients.elasticsearch.core.bulk.BulkOperation;
import co.elastic.clients.elasticsearch.core.search.HighlightField;
import co.elastic.clients.elasticsearch.core.search.Hit;
import co.elastic.clients.elasticsearch.indices.Alias;
import co.elastic.clients.elasticsearch.indices.CreateIndexRequest;
import co.elastic.clients.json.JsonData;
import co.elastic.clients.json.jackson.JacksonJsonpMapper;
import co.elastic.clients.transport.ElasticsearchTransport;
import co.elastic.clients.transport.endpoints.BooleanResponse;
import co.elastic.clients.transport.rest_client.RestClientTransport;

/**
 * 基于ElasticSearch-Java实现的Elastic客户服务
 * 
 * @author john.peng
 *
 */
public class ElasticUtilImpl implements ElasticUtil, ElasticLowLevelUtil {
	private static final Logger logger = LoggerFactory.getLogger(ElasticUtilImpl.class);

	private RestClient restClient = null;
	private ElasticsearchTransport transport = null;
	private ElasticsearchClient client = null;

	private ElasticConfigBean elasticConfigBean = null;
	private Map<String, String> indexes = new HashMap<>(), revIndexes = new HashMap<>();

	/**
	 * 构造函数
	 * @param elasticConfigBean 配置对象
	 */
	public ElasticUtilImpl(ElasticConfigBean elasticConfigBean) {
		super();
		this.elasticConfigBean = elasticConfigBean;
	}

	/**
	 * 添加一个缓存的索引
	 * 
	 * @param className 类名
	 * @param indexName 索引名
	 */
	protected void addCachedIndex(String className, String indexName) {
		indexes.put(className, indexName);
		revIndexes.put(indexName, className);
	}

	@Override
	public void destroy() throws Exception {
		client = null;
		if (transport != null) {
			transport.close();
			transport = null;
		}
		if (restClient != null) {
			restClient.close();
			restClient = null;
		}
	}

	@Override
	public void init() {
		long t0 = System.currentTimeMillis();
		if (logger.isInfoEnabled()) {
			logger.info("Initialize the elastic search client ......");
		}
		List<String> servers = elasticConfigBean.getServers();
		if (servers == null || servers.isEmpty()) {
			if (logger.isErrorEnabled()) {
				logger.error("Not define the elastic search server.");
			}
			throw new UserInterfaceSystemErrorException(SystemErrors.SYSTEM_ILLEGAL_PARAM);
		}
		HttpHost[] httpHosts = new HttpHost[servers.size()];
		for (int index = 0, num = httpHosts.length; index < num; index++) {
			httpHosts[index] = HttpHost.create(servers.get(index));
		}
		String certFile = elasticConfigBean.getSslCertFile();
		SSLContextBuilder sslContextBuilder = null;
		SSLContext sslContext = null;
		if (httpHosts[0].getSchemeName().equalsIgnoreCase("HTTPS")) {
			try (InputStream is = Files.newInputStream(Paths.get(certFile))) {
				CertificateFactory factory = CertificateFactory.getInstance("X.509");
				Certificate trustedCa = factory.generateCertificate(is);
				KeyStore trustStore = KeyStore.getInstance("pkcs12");
				trustStore.load(null, null);
				trustStore.setCertificateEntry("ca", trustedCa);
				sslContextBuilder = SSLContexts.custom().loadTrustMaterial(trustStore, null);
				sslContext = sslContextBuilder.build();
			} catch (IOException | CertificateException | KeyStoreException | NoSuchAlgorithmException
					| KeyManagementException ex) {
				if (logger.isErrorEnabled()) {
					logger.error(String.format("Load the CA ceritifcate fail."), ex);
				}
				throw new UserInterfaceSystemErrorException(SystemErrors.SYSTEM_ILLEGAL_PARAM);
			}
		}
		final CredentialsProvider credentialsProvider = new BasicCredentialsProvider();
		credentialsProvider.setCredentials(AuthScope.ANY,
				new UsernamePasswordCredentials(elasticConfigBean.getUser(), elasticConfigBean.getPassword()));
		final SSLContext sslContextFinal = sslContext;
		restClient = RestClient.builder(httpHosts).setHttpClientConfigCallback(ccb -> {
			if (sslContextFinal != null) {
				ccb.setSSLContext(sslContextFinal);
				if (elasticConfigBean.isIgnoreHostname()) {
					ccb.setSSLHostnameVerifier((hostname, session) -> true);
				}
			}
			return ccb.setDefaultIOReactorConfig(IOReactorConfig.custom().setIoThreadCount(1).build())
					.disableAuthCaching().setDefaultCredentialsProvider(credentialsProvider);
		}).setRequestConfigCallback(rcb -> {
			return rcb.setConnectTimeout(5000).setConnectionRequestTimeout(3000).setSocketTimeout(60000);
		}).build();
		transport = new RestClientTransport(restClient, new JacksonJsonpMapper());
		client = new ElasticsearchClient(transport);
		scanElasticEntitiesAndInitialize();
		if (logger.isInfoEnabled()) {
			logger.info(String.format("Initalize the elastic search client success, elapsed time: %d ms.",
					System.currentTimeMillis() - t0));
		}
	}

	@SuppressWarnings("unchecked")
	@Override
	public <T extends PO> boolean checkExist(T t) {
		String id = t.getId(), code = null;
		if (t instanceof BaseDict) {
			code = ((BaseDict) t).getCode();
		}
		if (t == null || !(StringUtils.isBlank(id) || StringUtils.isBlank(code))) {
			return false;
		}
		final String indexName = getIndex(t.getClass());
		boolean result = false;
		try {
			if (!result && !StringUtils.isBlank(id)) {
				BooleanResponse response = client.exists(erb -> erb.index(indexName).id(id));
				result = response != null && response.value();
			}
			if (!result && !StringUtils.isBlank(code)) {
				// 根据code判断是否存在
				final String _code = code;
				SearchResponse<T> res = client.search(
						srb -> srb.index(indexName).query(qb -> qb.match(mqb -> mqb.field("code").query(_code))),
						(Class<T>) t.getClass());
				result = res != null && res.hits() != null && res.hits().hits() != null && res.hits().hits().size() > 0;
			}
			return result;
		} catch (ElasticsearchException | IOException ex) {
			if (logger.isErrorEnabled()) {
				logger.error(String.format("", t.getClass().getName(), t.getId()), ex);
			}
			throw new UserInterfaceDalErrorException(DalErrors.ES_REST_FAIL);
		}
	}

	/**
	 * 返回ElaticsearchClient
	 * 
	 * @return ElasticsearchClient
	 */
	public ElasticsearchClient getElasticsearchClient() {
		return client;
	}

	/**
	 * 初始化扫描包路径下所有的Elastic Entity
	 */
	@SuppressWarnings("unchecked")
	private void scanElasticEntitiesAndInitialize() {
		List<String> packages = elasticConfigBean.getBasePackages();
		if (packages.size() > 0) {
			for (String p : packages) {
				ClassUtils.scanPackage(p, className -> {
					try {
						Class<? extends ElasticBaseEntity> clazz = (Class<? extends ElasticBaseEntity>) Class
								.forName(className);
						createIndex(clazz);
					} catch (Exception ex) {
						if (logger.isWarnEnabled()) {
							logger.warn(String.format("Initialize the index of the class[%s] fail.", className), ex);
						}
					}
				});
			}
		} else {
			if (logger.isWarnEnabled()) {
				logger.warn("There are not any elastic entity.");
			}
		}
	}

	@Override
	public List<NodeInfoBean> getAllNodes() {
		try {
			List<NodeInfoBean> list = new ArrayList<>();
			client.cat().nodes().valueBody().forEach(nr -> {
				NodeInfoBean nib = new NodeInfoBean();
				nib.setIp(nr.ip());
				nib.setCpu(Float.valueOf(nr.cpu()));
				nib.setHeapPercent(Float.valueOf(nr.heapPercent()));
				nib.setLoad_15m(Float.valueOf(nr.load15m()));
				nib.setLoad_1m(Float.valueOf(nr.load1m()));
				nib.setLoad_5m(Float.valueOf(nr.load5m()));
				nib.setMaster(Boolean.valueOf(nr.master()));
				nib.setName(nr.name());
				nib.setNodeRole(nr.nodeRole());
				nib.setRamPercent(Float.valueOf(nr.ramPercent()));
				list.add(nib);
			});
			return list;
		} catch (ElasticsearchException | IOException ex) {
			if (logger.isErrorEnabled()) {
				logger.error("Invoke ES CAT NODES API fail.", ex);
			}
			throw new UserInterfaceDalErrorException(DalErrors.ES_CAT_FAIL);
		}
	}

	@Override
	public List<IndicesInfoBean> getAllIndexes() {
		try {
			List<IndicesInfoBean> list = new ArrayList<>();
			client.cat().indices().valueBody().forEach(ir -> {
				IndicesInfoBean iib = new IndicesInfoBean();
				iib.setHealth(ir.health());
				iib.setDocDeleted(Long.valueOf(ir.docsDeleted()));
				iib.setDocsCount(Long.valueOf(ir.docsCount()));
				iib.setIndex(ir.index());
				iib.setPri(Integer.valueOf(ir.pri()));
				iib.setPriStoreSize(Long.valueOf(ir.priStoreSize()));
				iib.setRep(Integer.valueOf(ir.rep()));
				iib.setStatus(ir.status());
				iib.setStoreSize(Long.valueOf(ir.storeSize()));
				iib.setUuid(ir.uuid());
				list.add(iib);
			});
			return list;
		} catch (ElasticsearchException | IOException ex) {
			if (logger.isErrorEnabled()) {
				logger.error("Invoke ES CAT INDEX API fail.", ex);
			}
			throw new UserInterfaceDalErrorException(DalErrors.ES_CAT_FAIL);
		}
	}

	@Override
	public JSONObject getById(String id, String index) {
		try {
			GetRequest gReq = GetRequest.of(gb -> gb.index(index).id(id));
			return client.get(gReq, JSONObject.class).source();
		} catch (ElasticsearchException | IOException ex) {
			if (logger.isErrorEnabled()) {
				logger.error(String.format("Get the document[%s:%s] fail.", index, id), ex);
			}
			throw new UserInterfaceDalErrorException(DalErrors.ES_GET_FAIL);
		}
	}

	@Override
	public void deleteIndex(String index) {
		try {
			client.indices().delete(dir -> dir.index(index));
		} catch (ElasticsearchException | IOException ex) {
			if (logger.isErrorEnabled()) {
				logger.error(String.format("Delete the index[%s] fail.", index), ex);
			}
			throw new UserInterfaceDalErrorException(DalErrors.ES_INDEX_FAIL);
		}
	}

	@Override
	public void deleteIndex(String id, String index) {
		if (getById(id, index) != null) {
			try {
				client.delete(DeleteRequest.of(db -> db.index(index).id(id)));
				// 清理缓存
				revIndexes.remove(index);
			} catch (ElasticsearchException | IOException ex) {
				if (logger.isErrorEnabled()) {
					logger.error(String.format("Delete the index[%s:%s] fail.", index, id), ex);
				}
				throw new UserInterfaceDalErrorException(DalErrors.ES_DELETE_FAIL);
			}
		}
	}

	/**
	 * 扫描指定Elastic Entity类中的索引属性
	 *
	 * @param clazz  Elastic Entity类定义
	 * @param action 扫描到索引属性后执行的Action
	 */
	private void scanClassFields(Class<?> clazz, BiConsumer<String, ElasticField> action) {
		if (clazz.getName().equalsIgnoreCase("java.lang.Object")) {
			// 已经到了顶层对象，跳出
			return;
		}
		Field[] fields = clazz.getDeclaredFields();
		for (Field field : fields) {
			ElasticField annotationField = field.getAnnotation(ElasticField.class);
			if (annotationField != null) {
				action.accept(field.getName(), annotationField);
			}
		}
		scanClassFields(clazz.getSuperclass(), action);
	}

	private Property createProperty(ElasticField field) {
		Property.Kind type = field.type();
		boolean store = field.store();
		String analyzer = field.analyzer();
		String searchAnalyzer = field.searchAnalyzer();
		if (!StringUtils.isBlank(analyzer)) {
			// 为了预防可能配置了分词器，却忘记配置正确的类型，自动更正为"text"类型。
			type = Property.Kind.Text;
		}
		Property property = null;
		switch (type) {
		case Text:
			if (StringUtils.isBlank(analyzer)) {
				analyzer = "ik_max_word";
			}
			if (StringUtils.isBlank(searchAnalyzer)) {
				searchAnalyzer = "ik_smart";
			}
			final String _analyzer = analyzer, _searchAnalyzer = searchAnalyzer;
			property = Property.of(pb -> pb
					.text(kpb -> kpb.store(store).store(store).analyzer(_analyzer).searchAnalyzer(_searchAnalyzer)));
			break;
		case Boolean:
			property = Property.of(pb -> pb.boolean_(kpb -> kpb.store(store)));
			break;
		case Byte:
			property = Property.of(pb -> pb.byte_(kpb -> kpb.store(store)));
			break;
		case Completion:
			property = Property.of(pb -> pb.completion(kpb -> kpb.store(store)));
			break;
		case ConstantKeyword:
			property = Property.of(pb -> pb.constantKeyword(kpb -> kpb));
			break;
		case Date:
			property = Property.of(pb -> pb.date(kpb -> kpb.store(store)));
			break;
		case DateRange:
			property = Property.of(pb -> pb.dateRange(kpb -> kpb.store(store)));
			break;
		case Double:
			property = Property.of(pb -> pb.double_(kpb -> kpb.store(store)));
			break;
		case DoubleRange:
			property = Property.of(pb -> pb.doubleRange(kpb -> kpb.store(store)));
			break;
		case Float:
			property = Property.of(pb -> pb.float_(kpb -> kpb.store(store)));
			break;
		case FloatRange:
			property = Property.of(pb -> pb.floatRange(kpb -> kpb.store(store)));
			break;
		case GeoPoint:
			property = Property.of(pb -> pb.geoPoint(kpb -> kpb.store(store)));
			break;
		case GeoShape:
			property = Property.of(pb -> pb.geoShape(kpb -> kpb.store(store)));
			break;
		case Histogram:
			property = Property.of(pb -> pb.histogram(kpb -> kpb));
			break;
		case Integer:
			property = Property.of(pb -> pb.integer(kpb -> kpb.store(store)));
			break;
		case IntegerRange:
			property = Property.of(pb -> pb.integerRange(kpb -> kpb.store(store)));
			break;
		case Ip:
			property = Property.of(pb -> pb.ip(kpb -> kpb.store(store)));
			break;
		case IpRange:
			property = Property.of(pb -> pb.ipRange(kpb -> kpb.store(store)));
			break;
		case Long:
			property = Property.of(pb -> pb.long_(kpb -> kpb.store(store)));
			break;
		case LongRange:
			property = Property.of(pb -> pb.longRange(kpb -> kpb.store(store)));
			break;
		case Object:
			property = Property.of(pb -> pb.object(kpb -> kpb.store(store)));
			break;
		case Point:
			property = Property.of(pb -> pb.point(kpb -> kpb.store(store)));
			break;
		case Shape:
			property = Property.of(pb -> pb.shape(kpb -> kpb.store(store)));
			break;
		case Short:
			property = Property.of(pb -> pb.short_(kpb -> kpb.store(store)));
			break;
		case Keyword:
		default:
			property = Property.of(pb -> pb.keyword(kpb -> kpb.store(store)));
			break;
		}
		return property;
	}

	@Override
	public <T extends PO> void createIndex(Class<T> clazz) {
		ElasticIndex annotationIndex = clazz.getAnnotation(ElasticIndex.class);
		if (annotationIndex != null) {
			// 是一个Elastic实体类，需要处理索引
			try {
				String name = annotationIndex.value();
				if (StringUtils.isBlank(name)) {
					name = clazz.getName();
				}
				final String indexName = name;
				final Map<String, Property> mappings = new HashMap<>();
				// 处理fields
				scanClassFields(clazz, (fieldName, field) -> {
					if (!StringUtils.isBlank(field.value())) {
						fieldName = field.value();
					}
					Property property = createProperty(field);
					mappings.put(fieldName, property);
				});
				final Map<String, Alias> aliases = new HashMap<>();
				for (ElasticIndexAlias annotationAlias : annotationIndex.alias()) {
					Alias alias = Alias.of(aliasBuilder -> {
						aliasBuilder.indexRouting(annotationAlias.index_routing());
						aliasBuilder.isHidden(annotationAlias.is_hidden());
						aliasBuilder.isWriteIndex(annotationAlias.is_write_index());
						aliasBuilder.routing(annotationAlias.routing());
						aliasBuilder.searchRouting(annotationAlias.search_routing());
						if (annotationAlias.filter() != null) {
							// TODO 处理alias的过滤查询
						}
						return aliasBuilder;
					});
					aliases.put(annotationAlias.value(), alias);
				}
				if (!client.indices().exists(erb -> erb.index(indexName)).value()) {
					// 索引不存在
					CreateIndexRequest createIndexRequest = CreateIndexRequest.of(cirb -> {

						cirb.index(indexName);
						if (!mappings.isEmpty()) {
							cirb.mappings(tmb -> tmb.dynamic(DynamicMapping.False).properties(mappings));
						}
						// 处理alias
						if (aliases != null && !aliases.isEmpty()) {
							cirb.aliases(aliases);
						}
						// 处理Settings
						cirb.settings(sb -> {
							sb.numberOfReplicas(Integer.toString(annotationIndex.number_of_replicas()));
							sb.numberOfShards(Integer.toString(annotationIndex.number_of_shards()));
							sb.routingPartitionSize(annotationIndex.routing_partition_size());
							sb.autoExpandReplicas(Boolean.toString(annotationIndex.auto_expand_replicas()));
							return sb;
						});
						return cirb;
					});
					client.indices().create(createIndexRequest);
					addCachedIndex(clazz.getName(), createIndexRequest.index());
				}
			} catch (ElasticsearchException | IOException ex) {
				if (logger.isErrorEnabled()) {
					logger.error(String.format("Create the index[%s] fail.", clazz.getName()), ex);
				}
				throw new UserInterfaceDalErrorException(DalErrors.ES_INDEX_FAIL);
			}
		}
	}

	@Override
	public <T extends PO> void deleteIndex(Class<T> clazz) {
		ElasticIndex annotationIndex = clazz.getAnnotation(ElasticIndex.class);
		if (annotationIndex != null) {
			// 是一个Elastic实体类，需要处理索引
			String index = annotationIndex.value();
			if (StringUtils.isBlank(index)) {
				index = clazz.getName();
			}
			// 删除Index
			deleteIndex(index);
			// 清理缓存
			String className = clazz.getName();
			indexes.remove(className);
		}
	}

	@SuppressWarnings("unchecked")
	@Override
	public <T extends PO> void deleteIndex(List<T> ts) {
		if (ts == null || ts.isEmpty()) {
			if (logger.isWarnEnabled()) {
				logger.warn("The entity's list or id's list is null or empty.");
			}
			return;
		}
		List<BulkOperation> bulkList = new ArrayList<>(ts.size());
		for (int index = 0; index < ts.size(); index++) {
			Class<T> clazz = (Class<T>) ts.get(index).getClass();
			final String indexName = getIndex(clazz);
			String id = ts.get(index).getId();
			BulkOperation bo = BulkOperation.of(builder -> builder.delete(_do -> _do.index(indexName).id(id)));
			bulkList.add(bo);
		}
		try {
			client.bulk(br -> br.operations(bulkList));
		} catch (ElasticsearchException | IOException ex) {
			if (logger.isErrorEnabled()) {
				logger.error(String.format("Bulk delete entities fail, total: %d.", ts.size()), ex);
			}
			throw new UserInterfaceDalErrorException(UserInterfaceDalErrorException.DalErrors.ES_BULK_INDEX_FAIL);
		}

	}

	@Override
	public Class<?> getIndexClass(String index) {
		if (revIndexes.containsKey(index)) {
			String className = revIndexes.get(index);
			try {
				return Class.forName(className);
			} catch (ClassNotFoundException ex) {
				if (logger.isWarnEnabled()) {
					logger.error(String.format("The class[%s] not existed.", className), ex);
				}
				throw new UserInterfaceDalErrorException(UserInterfaceDalErrorException.DalErrors.ENTITY_INVALID_PO);
			}
		} else {
			throw new UserInterfaceDalErrorException(UserInterfaceDalErrorException.DalErrors.ENTITY_INDEX_NOT_FOUND);
		}
	}

	@Override
	public <T extends PO> List<T> search(ConditionGroup group, Class<? extends PO> clazz, Pagination pagination) {
		return search(group, null, Collections.singletonList(clazz), pagination);
	}

	private <T> List<T> search(Class<T> clazz, List<String> indices, List<Query> queries, List<SortOptions> orders,
			Pagination pagination) {
		try {
			SearchRequest sr = SearchRequest.of(sb -> {
				SearchRequest.Builder builder = sb.index(indices);
				Set<String> highlightFields = new HashSet<>();
				if (queries.size() > 1) {
					builder.query(QueryBuilders.bool().must(queries).build()._toQuery());
				} else {
					builder.query(queries.get(0));
				}
				if (orders != null && !orders.isEmpty()) {
					builder.sort(orders);
				}
				if (pagination != null) {
					int size = pagination.getSize(), page = pagination.getPage();
					if (size <= 0 || size > MAX_RECORD_NUM) {
						size = MAX_RECORD_NUM;
						pagination.setSize(size);
					}
					if (page <= 0) {
						page = 1;
						pagination.setPage(page);
					}
					builder.from((pagination.getPage() - 1) * size);
					builder.size(size);
				} else {
					// 默认设置最大记录
					builder.from(0);
					builder.size(MAX_RECORD_NUM);
				}
				highlightFields.forEach(field -> {
					builder.highlight(slb -> {
						slb.fields(field,
								HighlightField.of(aa -> aa.preTags("<span color='red'>").postTags("</span>")));
						return slb;
					});
				});
				return builder;
			});
			SearchResponse<T> response = client.search(sr, clazz);

			List<T> result = new ArrayList<>();
			if (pagination != null) {
				pagination.setTotal((int) response.hits().total().value());
			}

			if (response != null && response.hits() != null && response.hits().hits() != null
					&& response.hits().hits().size() > 0) {
				for (Hit<T> hit : response.hits().hits()) {
					T t = hit.source();
					((ElasticBaseEntity) t).setScore(hit.score());
					if (t instanceof ElasticGeoPointBaseEntity) {
						if (hit.sort() != null && !hit.sort().isEmpty()) {
							((ElasticGeoPointBaseEntity) t).setDistance(Double.valueOf(hit.sort().get(0)));
						}
					}
					if (hit.highlight() != null && hit.highlight().size() > 0) {
						// 注入高亮数据
						hit.highlight().forEach((k, v) -> {
							((ElasticBaseEntity) t).getHighlights().putAll(hit.highlight());
						});
					}
					result.add(t);
				}
			}
			if (logger.isDebugEnabled()) {
				logger.debug(String.format("Search successfully, total: %d.", result.size()));
			}
			return result;
		} catch (ElasticsearchException | IOException ex) {
			if (logger.isErrorEnabled()) {
				logger.error(String.format("Invoke search api fail, class: %s.", clazz.getName()), ex);

			}
			throw new UserInterfaceDalErrorException(DalErrors.ES_SEARCH_FAIL);
		}
	}

	@SuppressWarnings("unchecked")
	@Override
	public <T extends PO> List<T> search(ConditionGroup group, RecordOrderGroup orderGroup,
			List<Class<? extends PO>> classes, Pagination pagination) {
		List<String> indices = getIndices(classes);
		Class<T> clazz = (Class<T>) classes.get(0);
		Set<String> highlightFields = new HashSet<>();
		List<SortOptions> orders = new ArrayList<>();
		if (orderGroup != null && orderGroup.getOrders() != null && orderGroup.getOrders().size() > 0) {
			orderGroup.getOrders().forEach(ro -> {
				SortOrder so = (ro.getType() == OrderType.ASC ? SortOrder.Asc : SortOrder.Desc);
				FieldSort fs = FieldSort.of(fsb -> fsb.field(ro.getField()).order(so));
				orders.add(SortOptions.of(sob -> sob.field(fs)));
			});
		}
		return search(clazz, indices, Collections.singletonList(createQueryCondition(group, clazz, highlightFields)),
				orders, pagination);
	}

	private List<String> getIndices(List<Class<? extends PO>> classes) {
		List<String> indices = new ArrayList<>();
		if (classes == null || classes.isEmpty()) {
			indices.add("*");
		} else {
			classes.forEach(clazz -> indices.add(getIndex(clazz)));
		}
		return indices;
	}

	private FieldValue createFieldValue(Object value) {
		if (value instanceof Boolean) {
			return FieldValue.of((Boolean) value);
		} else if (value instanceof Long) {
			return FieldValue.of((Long) value);
		} else if (value instanceof Integer) {
			return FieldValue.of((Integer) value);
		} else if (value instanceof Short) {
			return FieldValue.of((Short) value);
		} else if (value instanceof Double) {
			return FieldValue.of((Double) value);
		} else if (value instanceof Float) {
			return FieldValue.of((Float) value);
		} else {
			return FieldValue.of((String) value);
		}
	}

	/**
	 * 判断Elastic Entity类中指定的属性是否定义为全文检索属性
	 *
	 * @param field 字段
	 * @param clazz Elastic Entity类定义
	 * @return 如果是全文检索（TEXT）类型返回true，否则返回false。
	 */
	private boolean isTextType(String field, Class<?> clazz) {
		if (clazz.getName().equalsIgnoreCase("java.lang.Object")) {
			return true;
		}
		try {
			ElasticField elasticField = clazz.getDeclaredField(field).getAnnotation(ElasticField.class);
			return (elasticField.type() == Property.Kind.Text);
		} catch (NoSuchFieldException ex) {
			return isTextType(field, clazz.getSuperclass());
		}
	}

	private Query createQueryCondition(ConditionGroup group, Class<?> clazz, Set<String> highlightFields) {
		if (group == null) {
			return QueryBuilders.matchAll().build()._toQuery();
		}
		BoolQuery.Builder bqb = QueryBuilders.bool();
		if (group.getItems().size() == 1 && group instanceof GeneralAccessor.ConditionTuple) {
			ConditionTuple tuple = (ConditionTuple) group;
			if (highlightFields != null) {
				highlightFields.add(tuple.field);
			}
			Query q = null;
			switch (tuple.operate) {
			case FUZZY:
				if (isTextType(tuple.field, clazz)) {
					// 已经有全文检索索引，使用TERM搜索，并根据值类型来确定使用TERMs还是TERM
					if (tuple.value instanceof Collection) {
						List<FieldValue> values = new ArrayList<>();
						((Collection<?>) tuple.value).forEach(v -> values.add(createFieldValue(v)));
						if (values.size() > 0) {
							q = QueryBuilders.terms().field(tuple.field).terms(tqb -> tqb.value(values)).build()
									._toQuery();
						} else {
							q = QueryBuilders.matchAll().build()._toQuery();
						}
					} else {
						q = QueryBuilders.term().field(tuple.field).value(createFieldValue(tuple.value)).build()
								._toQuery();
					}
				} else {
					// 无全文检索索引，使用正则表达式搜索
					if (tuple.value instanceof Collection) {
						List<Query> queries = new ArrayList<>();
						((Collection<?>) tuple.value).forEach(v -> queries.add(QueryBuilders.regexp().field(tuple.field)
								.value(String.format(".*(%s).*", v)).build()._toQuery()));
						if (queries.size() > 1) {
							q = QueryBuilders.bool().should(queries).build()._toQuery();
						} else if (queries.size() == 1) {
							q = queries.get(0);
						} else {
							q = QueryBuilders.matchAll().build()._toQuery();
						}
					} else {
						q = QueryBuilders.regexp().field(tuple.field).value(String.format(".*(%s).*", tuple.value))
								.build()._toQuery();
					}
				}
				break;
			case PREFIX:
				q = QueryBuilders.prefix().field(tuple.field).value((String) tuple.value).build()._toQuery();
				break;
			case EQ:
				q = QueryBuilders.match().field(tuple.field).query(createFieldValue(tuple.value)).build()._toQuery();
				break;
			case LT:
				q = QueryBuilders.range().field(tuple.field).lt(JsonData.of(tuple.value)).build()._toQuery();
				break;
			case GT:
				q = QueryBuilders.range().field(tuple.field).gt(JsonData.of(tuple.value)).build()._toQuery();
				break;
			case LTE:
				q = QueryBuilders.range().field(tuple.field).lte(JsonData.of(tuple.value)).build()._toQuery();
				break;
			case GTE:
				q = QueryBuilders.range().field(tuple.field).gte(JsonData.of(tuple.value)).build()._toQuery();
				break;
			case IS_NULL:
				q = QueryBuilders.bool().mustNot(QueryBuilders.exists().field(tuple.field).build()._toQuery()).build()
						._toQuery();
				break;
			case IS_NOT_NULL:
				q = QueryBuilders.exists().field(tuple.field).build()._toQuery();
				break;
			case IN:
				if (tuple.value instanceof Collection) {
					List<Query> queries = new ArrayList<>();
					((Collection<?>) tuple.value).forEach(v -> queries.add(
							QueryBuilders.match().field(tuple.field).query(createFieldValue(v)).build()._toQuery()));
					q = QueryBuilders.bool().should(queries).build()._toQuery();
				} else {
					q = QueryBuilders.match().field(tuple.field).query(createFieldValue(tuple.value)).build()
							._toQuery();
				}
				break;
			default:
				if (logger.isErrorEnabled()) {
					logger.error(String.format("Unsupported the operate type: %s.", tuple.operate));
				}
				throw new UserInterfaceSystemErrorException(
						UserInterfaceSystemErrorException.SystemErrors.SYSTEM_UNSUPPORTED_OPERATE);
			}
			return q;
		} else {
			List<Query> subQuery = new ArrayList<>();
			for (ConditionGroup subGroup : group.getItems()) {
				subQuery.add(createQueryCondition(subGroup, clazz, highlightFields));
			}
			if (group.getOperateType() == ConditionGroup.OperateType.AND) {
				bqb.must(subQuery);
			} else {
				bqb.should(subQuery);
			}
			return bqb.build()._toQuery();
		}
	}

	@Override
	public <T extends PO> long count(ConditionGroup group, List<Class<? extends PO>> classes) {
		try {
			Set<String> highlightFields = new HashSet<>();
			Query q = createQueryCondition(group, classes.get(0), highlightFields);
			return client.count(CountRequest.of(crb -> crb.index(getIndices(classes)).query(q))).count();
		} catch (ElasticsearchException | IOException e) {
			if (logger.isErrorEnabled()) {
				logger.error(String.format("Invoke the count[%s] fail.", classes.get(0).getName()));
			}
			throw new UserInterfaceDalErrorException(DalErrors.ES_INDEX_FAIL);
		}
	}

	@Override
	public <T extends ElasticGeoPointBaseEntity> List<T> geoNearBy(GeoPoint centerPoint, String field,
			double distanceMeters, ConditionGroup group, List<Class<? extends PO>> classes) {
		return geoNearBy(centerPoint, field, distanceMeters, group, classes, null);
	}

	@SuppressWarnings("unchecked")
	@Override
	public <T extends ElasticGeoPointBaseEntity> List<T> geoNearBy(GeoPoint centerPoint, String field,
			double distanceMeters, ConditionGroup group, List<Class<? extends PO>> classes, Pagination pagination) {
		List<String> indices = getIndices(classes);
		Class<T> clazz = (Class<T>) classes.get(0);
		Set<String> highlightFields = new HashSet<>();
		List<Query> queries = new ArrayList<>();
		if (group != null) {
			queries.add(createQueryCondition(group, classes.get(0), highlightFields));
		}
		GeoLocation geoLoc = GeoLocation.of(glb -> glb
				.latlon(LatLonGeoLocation.of(llglb -> llglb.lat(centerPoint.getLat()).lon(centerPoint.getLon()))));
		queries.add(QueryBuilders.geoDistance().field(field).location(geoLoc)
				.distance(String.format("%fm", distanceMeters)).distanceType(GeoDistanceType.Arc).build()._toQuery());
		GeoDistanceSort geoDistanceSort = GeoDistanceSort.of(gdsb -> gdsb.field(field).location(geoLoc)
				.distanceType(GeoDistanceType.Arc).unit(DistanceUnit.Meters).order(SortOrder.Asc));
		SortOptions geoOrder = SortOptions.of(sob -> sob.geoDistance(geoDistanceSort));
		return search(clazz, indices, queries, Collections.singletonList(geoOrder), pagination);
	}

	@Override
	public <T extends ElasticGeoPointBaseEntity> List<T> geoWithInPolygon(GeoShapePolygon polygon, String field,
			ConditionGroup group, GeoPoint centerPoint, List<Class<? extends PO>> classes) {
		return geoWithInPolygon(polygon, field, group, centerPoint, classes, null);
	}

	@SuppressWarnings("unchecked")
	@Override
	public <T extends ElasticGeoPointBaseEntity> List<T> geoWithInPolygon(GeoShapePolygon polygon, String field,
			ConditionGroup group, GeoPoint centerPoint, List<Class<? extends PO>> classes, Pagination pagination) {
		List<String> indices = getIndices(classes);
		Class<T> clazz = (Class<T>) classes.get(0);
		Set<String> highlightFields = new HashSet<>();
		List<Query> queries = new ArrayList<>();
		if (group != null) {
			queries.add(createQueryCondition(group, classes.get(0), highlightFields));
		}
		GeoShapeFieldQuery shapeQuery = GeoShapeFieldQuery
				.of(gsfqb -> gsfqb.relation(GeoShapeRelation.Within).shape(JsonData.of(polygon)));
		queries.add(QueryBuilders.geoShape().field(field).shape(shapeQuery).build()._toQuery());

		if (centerPoint != null) {
			GeoLocation geoLoc = GeoLocation.of(glb -> glb
					.latlon(LatLonGeoLocation.of(llglb -> llglb.lat(centerPoint.getLat()).lon(centerPoint.getLon()))));
			GeoDistanceSort geoDistanceSort = GeoDistanceSort.of(gdsb -> gdsb.field(field).location(geoLoc)
					.distanceType(GeoDistanceType.Arc).unit(DistanceUnit.Meters).order(SortOrder.Asc));
			SortOptions geoOrder = SortOptions.of(sob -> sob.geoDistance(geoDistanceSort));
			return search(clazz, indices, queries, Collections.singletonList(geoOrder), pagination);
		} else {
			return search(clazz, indices, queries, null, pagination);
		}
	}

	/**
	 * 获取指定Elastic Entity类对应的索引名
	 *
	 * @param clazz Elastic Entity类定义
	 * @return 索引名
	 */
	private String getIndex(Class<?> clazz) {
		String name = clazz.getName();
		if (indexes.containsKey(name)) {
			return indexes.get(name);
		} else {
			throw new UserInterfaceDalErrorException(UserInterfaceDalErrorException.DalErrors.ENTITY_INDEX_NOT_FOUND);
		}
	}

	@Override
	public <T extends PO> T getById(String id, Class<T> clazz) {
		if (StringUtils.isBlank(id)) {
			if (logger.isWarnEnabled()) {
				logger.warn("The field[id] is blank.");
			}
			throw new UserInterfaceSystemErrorException(SystemErrors.SYSTEM_ILLEGAL_PARAM);
		}
		final String index = getIndex(clazz);
		try {
			GetRequest gReq = GetRequest.of(gb -> gb.index(index).id(id));
			GetResponse<T> gResp = client.get(gReq, clazz);
			return gResp.source();
		} catch (ElasticsearchException | IOException ex) {
			if (logger.isErrorEnabled()) {
				logger.error(String.format("Get the document[%s:%s] fail.", index, id), ex);
			}
			throw new UserInterfaceDalErrorException(DalErrors.ES_GET_FAIL);
		}
	}

	@Override
	@SuppressWarnings("unchecked")
	public <T extends PO> T index(T t, boolean isNew) {
		if (t == null) {
			if (logger.isWarnEnabled()) {
				logger.warn("The entity is null.");
			}
			return null;
		}
		Class<T> clazz = (Class<T>) t.getClass();
		final String index = getIndex(clazz);
		String id = t.getId();
		if (StringUtils.isBlank(id)) {
			id = DigestUtils.uuid();
			t.setId(id);
		}
		final String _id = id;
		if (isNew) {
			try {
				IndexResponse response = client.index(IndexRequest.of(ir -> ir.index(index).id(_id).document(t)));
				return getById(response.id(), clazz);
			} catch (ElasticsearchException | IOException ex) {
				if (logger.isErrorEnabled()) {
					logger.error(String.format("Create index[%s:%s] fail.", clazz.getName(), _id), ex);
				}
				throw new UserInterfaceDalErrorException(DalErrors.ES_INDEX_FAIL);
			}
		} else {
			try {
				UpdateResponse<T> response = client.update(UpdateRequest.of(ur -> ur.index(index).id(_id).doc(t)),
						clazz);
				return getById(response.id(), clazz);
			} catch (ElasticsearchException | IOException ex) {
				if (logger.isErrorEnabled()) {
					logger.error(String.format("Update index[%s:%s] fail.", clazz.getName(), _id), ex);
				}
				throw new UserInterfaceDalErrorException(DalErrors.ES_INDEX_FAIL);
			}
		}
	}

	@Override
	@SuppressWarnings("unchecked")
	public <T extends PO> List<T> index(List<T> ts, List<Boolean> isNews) {
		if (ts == null || ts.isEmpty()) {
			if (logger.isWarnEnabled()) {
				logger.warn("The entity's list is null or empty.");
			}
			return ts;
		}
		List<BulkOperation> bulkList = new ArrayList<>(ts.size());
		for (int index = 0; index < ts.size(); index++) {
			Class<T> clazz = (Class<T>) ts.get(index).getClass();
			final String indexName = getIndex(clazz);
			T t = ts.get(index);
			String id = ts.get(index).getId();
			if (StringUtils.isBlank(id)) {
				id = DigestUtils.uuid();
				ts.get(index).setId(id);
			}
			boolean isNew = isNews.get(index);
			BulkOperation bo;
			final String _id = id;
			if (isNew) {
				bo = BulkOperation.of(builder -> builder.index(io -> io.index(indexName).id(_id).document(t)));
			} else {
				bo = BulkOperation
						.of(builder -> builder.update(uo -> uo.index(indexName).id(_id).action(ua -> ua.doc(t))));
			}
			bulkList.add(bo);
		}
		try {
			client.bulk(br -> br.operations(bulkList));
		} catch (ElasticsearchException | IOException ex) {
			if (logger.isErrorEnabled()) {
				logger.error(String.format("Bulk save entities fail, total: %d.", ts.size()), ex);
			}
			throw new UserInterfaceDalErrorException(UserInterfaceDalErrorException.DalErrors.ES_BULK_INDEX_FAIL);
		}
		return ts;
	}

}
